#include <iostream>
#include <string>
#include <semaphore.h>
#include <unistd.h>
#include <sstream>
#include <vector>
#include <thread>
#include <mutex>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/time.h>
#ifdef __linux__
#include <pty.h>
#include <sys/epoll.h>
#elif defined(__APPLE__) || defined(__MACH__)
#include <util.h>
#include <sys/event.h>
#else
#error "Unsupported platform"
#endif
#include "mqtt_config.h"
#include "jc-mtc.h"
#define _set_error_code         \
    {                           \
        _error_code = __LINE__; \
        mux.unlock();           \
        return -__LINE__;       \
    }
using namespace std;
#define var auto
bool _client_connected = false;
int _client_sock_fd = 0;
mutex mux;
int _error_code = 0;
bool _mqtt_client_connected = false;
vector<string> _cmd_queue;
vector<string> _ext_queue;

time_t get_current_time_ms()
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    return tv.tv_sec * 1000 + tv.tv_usec / 1000;
}

void on_message(char *topic, int topic_len, char *msg, int msg_len)
{
    if (topic_len > 4 && strncmp(topic + topic_len - 4, "/msg", 4) == 0)
    {
        if (_client_connected)
            send(_client_sock_fd, msg, msg_len, 0);
        return;
    }

    if (topic_len > 5 && strncmp(topic + topic_len - 5, "/pong", 5) == 0)
    {
        // show TTL there
        auto ms = get_current_time_ms();
        auto ttl = ms - stol(string(msg, msg_len));
        //  f'\x1b[0;32m TTL: {now_ms - float(msg)} ms \x1b[0m\r\n'
        string ttl_msg = "\x1b[0;32m TTL: " + to_string(ttl) + " ms \x1b[0m\r\n";
        if (_client_connected)
            send(_client_sock_fd, ttl_msg.c_str(), ttl_msg.length(), 0);
        return;
    }
}

void on_error(int error_code)
{
    if (!error_code)
    {
        _mqtt_client_connected = true;
        string time;
        // ping the current time
        auto ms = get_current_time_ms();
        time = to_string(ms);
        client_publish("ping", time);
        // send the command queue
        for (auto &cmd : _cmd_queue)
        {
            mqtt_out_raw(cmd);
        }
        // clear
        _cmd_queue.clear();
        for (auto &ext : _ext_queue)
        {
            client_publish("ext", ext);
        }
        // clear
        _ext_queue.clear();
        return;
    }
    // rasie an exception, and exit the master process
    _error_code = 1;
    _mqtt_client_connected = false;
    mux.unlock();
}
#pragma region telnet
/*
telnet commands:
      SE                  240    End of subnegotiation parameters.
      NOP                 241    No operation.
      Data Mark           242    The data stream portion of a Synch. This should always be accompanied by a TCP Urgent notification.
      Break               243    NVT character BRK.
      Interrupt Process   244    The function IP.
      Abort output        245    The function AO.
      Are You There       246    The function AYT.
      Erase character     247    The function EC.
      Erase Line          248    The function EL.
      Go ahead            249    The GA signal.
      SB                  250    Indicates that what follows is subnegotiation of the indicated option.
      WILL (option code)  251    Indicates the desire to begin performing, or confirmation that you are now performing, the indicated option.
      WON'T (option code) 252    Indicates the refusal to perform, or continue performing, the indicated option.
      DO (option code)    253    Indicates the request that the other party perform, or confirmation that you are expecting the other party to perform, the indicated option.
      DON'T (option code) 254    Indicates the demand that the other party stop performing, or confirmation that you are no longer expecting the other party to perform, the indicated option.
      IAC                 255    Data Byte 255.

*/
const unsigned char IAC = 255;             // \xff
const unsigned char SB = 250;              // \xfa
const unsigned char SE = 240;              // \xf0
const unsigned char NAWS = 31;             // \x1f
const unsigned char DO = 253;              // \xfd
const unsigned char DONT = 254;            // \xfe
const unsigned char WILL = 251;            // \xfb
const unsigned char WONT = 252;            // \xfc
const unsigned char _ECHO = 1;             // \x01
const unsigned char GA = 249;              // \xf9
const unsigned char Suppress_Go_Ahead = 3; // \x03
const unsigned char LINEMODE = 34;         // \x22

/*
ff fd 01: IAC WILL ECHO
ff fd 03: IAC WILL Suppress Go Ahead
>> start, after login
ff fb 25: IAC DO Authentication(37)
ff fd 03: IAC WILL Suppress Go Ahead(3)
ff fb 18: IAC DO Terminal Type(24)
ff fb 1f: IAC DO NAWS(31)
ff fb 20: IAC DO Terminal Speed(32)
ff fb 21: IAC DO Remote Flow Control(33)
ff fb 22: IAC DO Linemode(34)
ff fb 27: IAC DO New Environment Option(39)
ff fd 05: IAC WILL Status(5)
>> start NAWS negotiation
ff fa 1f 00 d9 00 0d ff f0: IAC SB NAWS 0 217 0 13 IAC SE
*/
void _telnet_send(unsigned char b1, unsigned char b2)
{
    unsigned char buf[3] = {IAC, b1, b2};
    send(_client_sock_fd, buf, 3, 0);
}
void _send_ext(string &ext)
{
    if (_mqtt_client_connected)
    {
        client_publish("ext", ext);
    }
    else
    {
        _ext_queue.push_back(ext);
    }
}
int go_WILL(unsigned char *b, int len)
{
    string msg;
    switch (b[0])
    {
    case NAWS:
        _telnet_send(DO, NAWS);
        return 2;
    case _ECHO:
        msg = "echo";
        _send_ext(msg);
        _telnet_send(DO, _ECHO);
        return 2;
    case Suppress_Go_Ahead:
        _telnet_send(DO, Suppress_Go_Ahead);
        return 2;
    default:
        return 2;
    }
}
int go_WONT(unsigned char *b, int len)
{
    string msg;
    switch (b[0])
    {
    case NAWS:
        _telnet_send(DONT, NAWS);
        return 2;
    case _ECHO:
        msg = "-echo";
        _send_ext(msg);
        _telnet_send(DONT, _ECHO);
        return 2;
    case Suppress_Go_Ahead:
        _telnet_send(DONT, Suppress_Go_Ahead);
        return 2;
    default:
        return 2;
    }
}
int go_DO(unsigned char *b, int len)
{
    string msg;
    switch (b[0])
    {
    case NAWS:
        _telnet_send(WILL, NAWS);
        return 2;
    case _ECHO:
        msg = "echo";
        _send_ext(msg);
        return 2;
    case Suppress_Go_Ahead:
        _telnet_send(WILL, Suppress_Go_Ahead);
        return 2;
    default:
        return 2;
    }
}
int go_DONT(unsigned char *b, int len)
{
    string msg;
    switch (b[0])
    {
    case NAWS:
        _telnet_send(WONT, NAWS);
        return 2;
    case _ECHO:
        msg = "-echo";
        _send_ext(msg);
        return 2;
    case Suppress_Go_Ahead:
        _telnet_send(WONT, Suppress_Go_Ahead);
        return 2;
    default:
        return 2;
    }
}
int go_SB(unsigned char *b, int len)
{
    if (b[0] == NAWS)
    {
        // NAWS
        if (len < 7)
            return 1;
        short cols = b[1] * 256 + b[2];
        short rows = b[3] * 256 + b[4];
        string msg = "size " + to_string(rows) + " " + to_string(cols);
        _send_ext(msg);
        return 8;
    }
    else
    {
        // skip the subnegotiation
        int i = 1;
        for (; i < len; i++)
        {
            if (b[i] == IAC && b[i + 1] == SE)
                break;
        }
        if (i >= len)
            return 1;
        return i + 2;
    }
}

void _filter_data_from_sock(char *data, int len)
{
    static unsigned char _buf[1024 * 3];
    int i = 0, j = 0;
    unsigned char *b = (unsigned char *)data;
    for (i = 0; i < len; i++)
    {
        if (b[i] != IAC)
        {
            _buf[j++] = b[i];
            continue;
        }
        i++;
        if (i >= len)
            break;
        if (b[i] == IAC)
        {
            _buf[j++] = b[i];
            continue;
        }
        if (b[i] == WILL)
        {
            i += go_WILL(b + i + 1, len - i - 1);
            i--;
            continue;
        }
        if (b[i] == WONT)
        {
            i += go_WONT(b + i + 1, len - i - 1);
            i--;
            continue;
        }
        if (b[i] == DO)
        {
            i += go_DO(b + i + 1, len - i - 1);
            i--;
            continue;
        }
        if (b[i] == DONT)
        {
            i += go_DONT(b + i + 1, len - i - 1);
            i--;
            continue;
        }
        if (b[i] == SB)
        {
            i += go_SB(b + i + 1, len - i - 1);
            i--;
            continue;
        }
    }

    if (j)
    {
        if (_mqtt_client_connected)
            mqtt_out_raw((char *)_buf, j);
        else
        {
            // push to _cmd_queue, and public when mqtt connected
            _cmd_queue.push_back(string((char *)_buf, j));
        }
    }
}
#pragma endregion
// read from the socket, and send the data to the mqtt broker
#if __linux__
int _read_from_socket(int fd)
{
    int epoll_fd = epoll_create1(0);
    if (epoll_fd == -1)
    {
        _set_error_code;
    }
    struct epoll_event event;
    event.events = EPOLLIN;
    event.data.fd = fd;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1)
    {
        _set_error_code;
    }
    const int MAX_EVENTS = 8;
    struct epoll_event events[MAX_EVENTS];
    const int buf_len = 1024 * 3;
    static char buf[buf_len];
    while (1)
    {
        int num_events = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (num_events == -1)
        {
            _set_error_code;
        }
        for (int i = 0; i < num_events; i++)
        {
            if (events[i].events & EPOLLIN)
            {

                int num_read = read(events[i].data.fd, buf, buf_len);
                if (num_read == -1)
                {
                    close(epoll_fd);
                    _set_error_code;
                }
                if (num_read == 0)
                {
                    if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, NULL) == -1)
                    {
                        close(epoll_fd);
                        _set_error_code;
                    }
                }
                else
                {
                    if (_client_connected)
                        _filter_data_from_sock(buf, num_read);
                }
            }
        }
    }
    close(epoll_fd);
    _error_code = 1;
    mux.unlock();
    return 0;
}
#else
int _read_from_socket(int fd)
{
    int kq = kqueue();
    if (kq == -1)
    {
        _set_error_code;
    }
    struct kevent event;
    EV_SET(&event, fd, EVFILT_READ, EV_ADD, 0, 0, NULL);

    if (kevent(kq, &event, 1, NULL, 0, NULL) == -1)
    {
        _set_error_code;
    }
    const int MAX_EVENTS = 8;
    struct kevent events[MAX_EVENTS];
    const int buf_len = 1024 * 3;
    static char buf[buf_len];
    while (1)
    {
        int num_events = kevent(kq, NULL, 0, events, MAX_EVENTS, NULL);
        if (num_events == -1)
        {
            break;
        }
        for (int i = 0; i < num_events; i++)
        {
            if (events[i].filter == EVFILT_READ)
            {
                int num_read = read(events[i].ident, buf, buf_len);
                if (num_read == -1)
                {
                    close(kq);
                    _set_error_code;
                }
                if (num_read == 0)
                {
                    close(kq);
                    _error_code = 0;
                    mux.unlock();
                    return 0;
                }
                else
                {
                    if (_client_connected)
                        _filter_data_from_sock(buf, num_read);
                }
            }
        }
    }
    close(kq);
    _error_code = 1;
    mux.unlock();
    return 0;
}
#endif

int main(int argn, char *argv[])
{
    int server_fd, valread;
    struct sockaddr_in address;
    int opt = 1;
    int addrlen = sizeof(address);
    char buffer[1024] = {0};
    int listen_port = 23;
    int _n = 1;
    if (argn > _n && isdigit(argv[1][0]))
    {
        listen_port = atoi(argv[1]);
        _n++;
    }
    string confg_file = "../mqtt.ini";
    if (argn > _n)
        confg_file = argv[_n];
    int rc = 0;
    rc = read_config(confg_file);
    if (rc)
        exit(rc);
    server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_fd <= 0)
    {
        perror("socket failed");
        exit(__LINE__);
    }
#if __linux__
    // Forcefully attaching socket to the port
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)) == -1)
    {
        perror("setsockopt");
        exit(__LINE__);
    }
#endif
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(listen_port);

    if (::bind(server_fd, (struct sockaddr *)&address,
               sizeof(address)) < 0)
    {
        perror("bind failed");
        exit(__LINE__);
    }
    if (listen(server_fd, 1) < 0)
    {
        perror("listen");
        exit(__LINE__);
    }
    thread t1;
    set_on_message_callback(on_message);
    set_on_error_callback(on_error);
    try
    {
        while (1)
        {
            if ((_client_sock_fd = accept(server_fd, (struct sockaddr *)&address,
                                          (socklen_t *)&addrlen)) < 0)
            {
                perror("accept");
                close(server_fd);
                exit(__LINE__);
            }
#ifdef DEBUG
            cout << "client " << inet_ntoa(address.sin_addr) << ":" << ntohs(address.sin_port) << " connected" << endl;
#endif
            _client_connected = true;
            mux.lock();
            start_loop();
// create a thread to listen to the socket
#if __linux__
            t1 = thread(_read_from_socket, _client_sock_fd);
#else
            t1 = thread([]()
                        { _read_from_socket(_client_sock_fd); });
#endif
            t1.detach();
            // wait the mutex to be released
            mux.lock();
            if (_error_code)
            {
                throw std::runtime_error("error in the thread:" + to_string(_error_code));
            }

            // after the aquire the mux, close the socket and mqtt client
            _client_connected = false;
            close(_client_sock_fd);
            string msg = "exit";
            client_publish("ext", msg); // notify to close pty
            stop_loop();
            _mqtt_client_connected = false;
            if (t1.joinable())
                t1.join();
            mux.unlock();
#ifdef DEBUG
            cout << "client disconnected" << endl;
#endif
        }
    }
    catch (const std::exception &e)
    {
        std::cerr << e.what() << '\n';
    }
    if (_client_connected)
    {
        _client_connected = false;
        close(_client_sock_fd);
        stop_loop();
        if (t1.joinable())
            t1.join();
    }
    close(server_fd);
    mux.unlock();
    return 0;
}